package com.jida.hadoop.mr.topn;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util. *;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.jida.hadoop.mr.tools.StringComparator;
import com.jida.hadoop.mr.tools.TMatcher;
import com.jida.hadoop.mr.tools.TProperties;

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
/** 
	 * @包名   mr_job.dx 
	 * @文件名 DxTopN.java 
	 * @版本 V 1.0 
	 * @功能描述 用户行为取数据topN
	 */
public class DxTopN {

	/**
	 * @param 
	 * args[0]：输入路径（个人地址统计结果数据）
	 * args[1]：结果输出路径
	 * args[2]：排序字段（用户id）=1
	 * args[3]：排序值（访问次数）=2
	 * args[4]：top取值最大范围   =5
	 * Description：mapreduce统计用户访问热门地址，取前N条记录
	 * */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		//设置map和reduce内存使用大小
		conf.set("mapreduce.map.memory.mb", "3072");
		conf.set("mapreduce.reduce.memory.mb", "2048");
		//参数传递
		//排序字段：用户ID
		conf.set("topkey", "1");
		//排序值：PV访问量
		conf.set("topvalue", "2");
		//排序取前N条记录：前五条
		conf.set("topn", "5");
		Job job =Job.getInstance(conf, "Dx_TopN");
		job.setJarByClass(DxTopN.class);
		//map与reduce设置
		job.setMapperClass(TopNMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setReducerClass(TopNReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		//输入路径
		//FileInputFormat.addInputPath(job, new Path("file:///D:/测试数据/dx/03.DxFileUser/part-r-*"));
		FileInputFormat.addInputPath(job, new Path("file:///D:/测试数据/dx/DxStaOne/part-r-*"));
		//输出路径
		FileOutputFormat.setOutputPath(job, new Path("file:///D:/测试数据/dx/10.DxTopN"));
		//提交作业 判断退出条件（0正常退出，1非正常退出）
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

	/**
	 * 按指定字段作为key分组
	 * */
	public static class TopNMapper extends Mapper<LongWritable, Text, Text, Text> {
		private String topkey ;
		private Text okey = new Text();
		//获取分组字段位置
		@Override
		protected void setup( Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
			topkey = context.getConfiguration().get("topkey");
		}

		//以分组字段为key
		@Override
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] values = value.toString().split(TProperties.getValue("fileoutsplit"));
			okey = new Text(values[Integer.parseInt(topkey)]);
			//按用户分组
			context.write(okey, value);
		}
	}

	/**
	 * 排序取topN条记录
	 * */
	public static class TopNReducer extends Reducer<Text, Text, NullWritable, Text> {
		private Text result = new Text();
		private int topvalue ;
		private int topn ;
		//获取参数
		@Override
		protected void setup( Reducer<Text, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {
			//排序值：访问次数
			topvalue = Integer.parseInt(context.getConfiguration().get("topvalue"));
			//取N值：默认值为5
			topn = Integer.parseInt(context.getConfiguration().get("topn"));
		}

		//reduce排序取值
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
			TreeMap<Integer, String> topMap = new TreeMap<Integer, String>();
			//循环数据
		    for (Text val : values) {
			String[] str = val.toString().split(TProperties.getValue("fileoutsplit"));
			//将数据放入数据集合
			topMap.put(Integer.parseInt(str[topvalue]), val.toString());
			//移除超过取值条数的较小数据
			if(topMap.size() > topn){
				topMap.remove(topMap.firstKey());
			}
		    }
		    int num = 0;
		    Iterator<Integer> iterator = topMap.keySet().iterator();
		    //数据转换，计算排名
		    List<String> list = new ArrayList<String>();
		    while (iterator.hasNext()) {
			list.add(topMap.get(iterator.next()));
		    }
		    //取数据条数+1，用于计算排名
		    num = list.size() + 1 ;
		    for(int i = 0 ;i< list.size() ; i++){
			//排名
			num = num -1 ;
			//行为ID，用户ID，PV（访问次数），排名
			result = new Text(list.get(i) + TProperties.getValue("outfilesplit") + num );
				context.write(NullWritable.get(), result);
		    }
		}
	}
}